/*
 * Copyright 2011-2012 by The Regents of the University of California Licensed
 * under the Apache License, Version 2.0 (the "License"); you may not use this
 * file except in compliance with the License. you may obtain a copy of the
 * License from
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package edu.ucsd.db.bassarisk.storage.loading;

import java.nio.ByteBuffer;

import edu.ucsd.db.bassarisk.data.compression.VariableLengthUnsignedLongSequenceCodec;
import edu.ucsd.db.bassarisk.storage.common.ColumnStore;

public abstract class VariableLengthLongSegmentIndexer implements
    IColumnSegmentIndexer {
  protected final int COMMON_HEADER_BYTES = 3;

  protected final VariableLengthUnsignedLongSequenceCodec longCodec;

  protected final ByteBuffer entryRangeIndexBuffer;
  protected int numSegmentsIndexed;
  protected int entryCount;
  protected int bytesConsumed;

  public VariableLengthLongSegmentIndexer() {
    super();
    this.longCodec = VariableLengthUnsignedLongSequenceCodec.INSTANCE;

    this.entryRangeIndexBuffer = ByteBuffer.allocate(ColumnStore.BYTES_IN_DRAWER);
  }

  @Override
  public int cardsRequired() {
    double cardsRequired = ((double) this.bytesConsumed) / ColumnStore.BYTES_IN_CARD;
    return (int) Math.ceil(cardsRequired);
  }

  @Override
  public void reset() {
    this.numSegmentsIndexed = 0;
    this.entryCount = 0;
    this.entryRangeIndexBuffer.clear();
    this.bytesConsumed = this.COMMON_HEADER_BYTES;
  }

  protected int writeCommonData(final ByteBuffer drawer, final int firstCard) {
    int cardsRequired = this.cardsRequired();
    // Fail if there is not enough space
    assert (firstCard + cardsRequired) <= ColumnStore.CARDS_IN_DRAWER;

    int entryStartOffset = firstCard * ColumnStore.BYTES_IN_CARD;
    System.arraycopy(this.entryRangeIndexBuffer.array(), 0, drawer.array(),
                     entryStartOffset, this.entryRangeIndexBuffer.position());
    return this.entryRangeIndexBuffer.position();
  }

  protected int writeCommonHeaders(final ByteBuffer drawer) {
    int indexCardsUsed = this.cardsRequired();
    assert indexCardsUsed < 256;
    assert this.numSegmentsIndexed < 65513;
    drawer.put((byte) indexCardsUsed).putShort((short) this.numSegmentsIndexed);
    return this.COMMON_HEADER_BYTES;
  }
}